package com.tianji.remark.service.impl;

import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.tianji.api.dto.remark.LikedTimesDTO;
import com.tianji.common.autoconfigure.mq.RabbitMqHelper;
import com.tianji.common.constants.MqConstants;
import com.tianji.common.utils.CollUtils;
import com.tianji.common.utils.StringUtils;
import com.tianji.common.utils.UserContext;
import com.tianji.remark.constants.RedisConstants;
import com.tianji.remark.domain.dto.LikeRecordFormDTO;
import com.tianji.remark.domain.po.LikedRecord;
import com.tianji.remark.mapper.LikedRecordMapper;
import com.tianji.remark.service.ILikedRecordService;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.dao.DataAccessException;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.connection.StringRedisConnection;
import org.springframework.data.redis.core.Cursor;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.ScanOptions;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.ZSetOperations;
import org.springframework.stereotype.Service;

import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.stream.Collectors;
import java.util.stream.IntStream;

/**
 * <p>
 * 点赞记录表 服务实现类
 * </p>
 *
 * @author yb0os1
 * @since 2024-12-27
 */
@Slf4j
@Service
@RequiredArgsConstructor
public class LikedRecordRedisServiceImpl extends ServiceImpl<LikedRecordMapper, LikedRecord> implements ILikedRecordService {

    private final RabbitMqHelper rabbitMqHelper;
    private final StringRedisTemplate redisTemplate;


    @Override
    public void likedOrCancel(LikeRecordFormDTO dto) {
        //1、获取用户id
        Long userId = UserContext.getUser();
        String key = RedisConstants.LIKES_BIZ_KEY_PREFIX + dto.getBizId();
        //点赞或者取消点赞是否成功 true代表处理成功 false 失败
        boolean flag = dto.getLiked() ? likedHandle(userId, key) : unlikedHandle(userId, key);
        if (!flag) {
            log.error("点赞or取消点赞失败");
            return;
        }
        //4、统计redis中点赞数 并且缓存到redis
        Long likedTimes = redisTemplate.opsForSet().size(key);
        if (likedTimes == null) {
            return;
        }
        redisTemplate.opsForZSet()
                .add(RedisConstants.LIKE_BIZ_TYPE_PREFIX + dto.getBizType(), dto.getBizId().toString(), likedTimes);
    }

    @Override
    public Set<Long> batchQueryLikedStatus(List<Long> ids) {
        //使用redis管道技术
        List<Object> objects = redisTemplate.executePipelined((RedisCallback<Object>) connection -> {
            StringRedisConnection src = (StringRedisConnection) connection;
            Long userId = UserContext.getUser();
            for (Long id : ids) {
                String key = RedisConstants.LIKES_BIZ_KEY_PREFIX + id;
                src.sIsMember(key, userId.toString());//属于对应true 不属于对应false
            }
            return null;
        });
/*        Set<Long> res = new HashSet<>();
        for (int i = 0; i < objects.size(); i++) {
            Boolean b = (Boolean)objects.get(i);
            if (b){
                //如果ids中对应的角标属于redis中
                res.add(ids.get(i));
            }
        }
        return res;*/
        //等价于
        return IntStream.range(0, objects.size())//创建从0到集合size的流
                .filter(i -> (Boolean) objects.get(i))//遍历每一个元素 保留结果为true的角标i
                .mapToObj(ids::get)//将角标i转换为对应的id 也就是点赞过的id
                .collect(Collectors.toSet());
    }

    //将redis的点赞数量同步到mysql
    @Override
    public void readLikedTimesAndSendMessage(String bizType, int maxBizSize) {
        if (StringUtils.isBlank(bizType)) {
            return;
        }
        String key = RedisConstants.LIKE_BIZ_TYPE_PREFIX + bizType;
        //取出点赞最少的三十条数据 并从redis中删除
        Set<ZSetOperations.TypedTuple<String>> tuples = redisTemplate.opsForZSet().popMin(key, maxBizSize);
        if (CollUtils.isEmpty(tuples)) {
            return;
        }
        List<LikedTimesDTO> list = new ArrayList<>(tuples.size());
        for (ZSetOperations.TypedTuple<String> tuple : tuples) {
            String bizId = tuple.getValue();
            Double likedTimes = tuple.getScore();
            if (likedTimes == null || bizId == null) {
                continue;
            }
            list.add(LikedTimesDTO.of(Long.parseLong(bizId), likedTimes.intValue()));
        }
        //发送消息 提醒批量更新mysql
        if (CollUtils.isNotEmpty(list)) {
            log.info("发送消息 提醒批量更新mysql");
            rabbitMqHelper.send(
                    MqConstants.Exchange.LIKE_RECORD_EXCHANGE,
                    StringUtils.format(MqConstants.Key.LIKED_TIMES_KEY_TEMPLATE, bizType),
                    list
            );
        }
    }


    //将redis的点赞记录同步到mysql
/*    @Override
    public void readLikedRecordAndSendMessage() {
        //1、读取主键 likes:bizId:* 点赞记录的key
        Set<String> bizIds = scan(RedisConstants.LIKES_BIZ_KEY_PREFIX + "*");
        //2、建立Map<Long,Set<String>> bizId -> userIdSet
        Map<Long, Set<String>> map = bizIds.stream()
                .map(bizId -> bizId.substring(RedisConstants.LIKES_BIZ_KEY_PREFIX.length()))
                .map(Long::parseLong)
                .collect(Collectors.toMap(
                        bizId -> bizId,
                        bizId -> redisTemplate.opsForSet().members(RedisConstants.LIKES_BIZ_KEY_PREFIX + bizId)
                ));
        log.info("bizId -> userIdSet{}", map);
        //3、向MQ发送消息（如果消息过多可以分批次发送）
        rabbitMqHelper.send(
                MqConstants.Exchange.LIKE_RECORD_EXCHANGE,
                MqConstants.Key.QA_RECORD_TIMES_KEY,
                map
        );
    }*/

    private boolean unlikedHandle(Long userId, String key) {
        // 执行SREM命令
        Long removed = redisTemplate.opsForSet().remove(key, String.valueOf(userId));
        return removed != null && removed > 0;
    }

    private boolean likedHandle(Long userId, String key) {
        //1、 bizid:1 userId 向redis的set结果插入
        Long isAdd = redisTemplate.opsForSet().add(key, String.valueOf(userId));
        //2、返回结果
        return isAdd != null && isAdd > 0;
    }

    /**
     * 扫描主键，建议使用
     *
     * @param patten
     * @return
     */
    private Set<String> scan(String patten) {
        Set<String> keys = redisTemplate.execute((RedisCallback<Set<String>>) connection -> {
            Set<String> result = new HashSet<>();
            try (Cursor<byte[]> cursor = connection.scan(ScanOptions.scanOptions()
                    .match(patten).count(10000).build())) {
                while (cursor.hasNext()) {
                    result.add(new String(cursor.next()));
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
            return result;
        });
        return keys;
    }
}
